package com.cheetah.client;

import com.cheetah.config.ClientConfig;
import com.cheetah.constants.ProtocolConstants;
import com.cheetah.handler.ClientHandlerInitalizer;
import com.cheetah.listener.Callback;
import com.cheetah.listener.CheetahCallback;
import com.cheetah.protocol.CheetahMessage;
import com.cheetah.protostuff.ProtostuffSerialization;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * @Author Administrator
 * @Date 2019/7/22
 * @Version V1.0
 * @Description:
 **/
@Component
@Slf4j
public class CheetahClient<R,T> {

    private volatile Channel channel;

    @Autowired
    private ClientConfig clientConfig;

    @Autowired
    private ClientHandlerInitalizer clientHandlerInitalizer;
    EventLoopGroup workerGroup = new NioEventLoopGroup();
    Bootstrap bootstrap = new Bootstrap();
    @PostConstruct
    public void init(){

        bootstrap.group(workerGroup);
        //指定所使用的NIO传输channel
        bootstrap.channel(NioSocketChannel.class);
        //指定客户端初始化处理
        bootstrap.handler(clientHandlerInitalizer);

        new Thread(() -> doConnect()).start();
    }

    public Channel getClient(){

        if(channel != null && channel.isActive())
            return channel;
        return doConnect();
    }

    private synchronized Channel doConnect() {
        if (channel != null && channel.isActive()) {
            return channel;
        }

        log.info("connect to server {} on port {}",clientConfig.getIp(),clientConfig.getPort());
        ChannelFuture future = bootstrap.connect(clientConfig.getIp(), clientConfig.getPort());

        future.addListener(new ChannelFutureListener() {
            public void operationComplete(ChannelFuture futureListener) throws Exception {
                log.info("connect server status {}",future.isSuccess());
                if (!future.isSuccess()) {
                    future.channel().pipeline().fireChannelInactive();
                }
                channel = future.channel();
//                if (futureListener.isSuccess()) {
//                    channel = futureListener.channel();
//                    log.info("Connect to server successfully!");
//                } else {
//                    log.info("Failed to connect to server, try connect after 10s");
//
//                    futureListener.channel().eventLoop().schedule(new Runnable() {
//                        @Override
//                        public void run() {
//                            doConnect();
//                        }
//                    }, 10, TimeUnit.SECONDS);
//                }
            }
        });

        return future.channel();
    }

    public T sendMessage(R t, String type){
        Callback<T> callback = new CheetahCallback<>();

        byte[] data = ProtostuffSerialization.serialize(t);

        CheetahMessage cheetahMessage = CheetahMessage.builder()
                .magic(ProtocolConstants.PROTOCOL_MAGIC_PROTO)
                .version(ProtocolConstants.PROTOCOL_VERSION_PROTO)
                .type(type)
                .flag(ProtocolConstants.PROTOCOL_FlAG_PROTO_0)
//                .headLen(0)
                .bodyLen(data.length)
                .data(data)
                .build();

        getClient().writeAndFlush(cheetahMessage);
        getClient().attr(AttributeKey.valueOf("a")).set(callback);
        callback.start();
        return callback.getData();
    }

}
